package com.juma.cartridge.retryque;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import lombok.extern.slf4j.Slf4j;

import com.juma.cartridge.retryque.cmd.Command;
import com.juma.cartridge.retryque.cmd.CommandWorker;
import com.juma.cartridge.retryque.dao.RetryQueueDao;
/**
 * 	重试任务队例管理器
 * @ClassName TaskQueueManager
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年6月19日 上午10:07:54
 * @Version 1.0
 *
 */
@Slf4j
public class DefaultRetryQueueManager implements RetryQueueManager {

	@Resource
	private RetryQueueDao retryQueueDao;
	private Conf conf;
	private Map<String,RetryHandler> retryHandlers;
	private ScheduledFuture<?> taskScanner;
	
	private ScheduledExecutorService scannerExecutor;
	private ExecutorService retryExecutor;
	private ThreadGroup commandExecutors;
	private Thread keepperExecutor;
	
	
	private DelayQueue<TaskLeaseProlong> leaseQueue;
	private BlockingQueue<Command> commandQue;
	private LeaseProlongKeepper keepper;
	private TaskScanner scanner;
	
	private int commandWorkerCount = 3;
	
	public DefaultRetryQueueManager(){
		conf = new Conf();
		retryHandlers = new ConcurrentHashMap<String,RetryHandler>();
		leaseQueue = new DelayQueue<TaskLeaseProlong>();
	}
	
	public void initQueue() {
		if(scannerExecutor!=null) {
			return;
		}
		synchronized (this) {
			if (scannerExecutor==null) {
			    //初始化定时扫描线程池
				scannerExecutor = Executors.newScheduledThreadPool(1,new RetryThreadFactory("scanner"));
				//初始化任务补偿线程池
				int cores = conf.getLockCount();
				retryExecutor = buildThreadPool("retryWorker",cores,cores*2,new ArrayBlockingQueue<Runnable>(cores*3));
				
				//初始化任务状态回写线程组
				commandExecutors = new ThreadGroup("commandExecutors");
				
				commandQue = new ArrayBlockingQueue<Command>(conf.getLockCount()*2*4);
				scanner = new TaskScanner(conf,leaseQueue,retryQueueDao,retryHandlers,retryExecutor,commandQue);
				//启动定时扫描
				taskScanner=scannerExecutor.scheduleAtFixedRate(scanner, 5, conf.getLoopPeriod(), TimeUnit.SECONDS);
				
				//启动续租线程
				keepper = new LeaseProlongKeepper(leaseQueue,retryQueueDao);
				keepperExecutor = new Thread(keepper,"RetryQue-keepper");
				keepperExecutor.start();
				
				//启动命令工作线程
				for(int i=0;i<commandWorkerCount;i++) {
				    CommandWorker cw = new CommandWorker(conf,commandQue,retryQueueDao);
				    Thread th = new Thread(commandExecutors,cw,"RetryQue-commandWorker-"+i);
				    th.start();
				}
				log.info("RetryQue已启动...");
			}
		}
		
	}
	
	
	/**
	 * 创建业务重试线程池
	 * @return
	 */
	private ThreadPoolExecutor buildThreadPool(String name,int cores,int max,BlockingQueue<Runnable> que) {
	    ThreadPoolExecutor pe = new ThreadPoolExecutor(cores, max,
            60L, TimeUnit.SECONDS,que,new RetryThreadFactory(name));
	    return pe;
	}
	
	
	public void destoryQueue() {
		if(scannerExecutor!=null) {
			keepper.close();
			scannerExecutor.shutdownNow();
			retryExecutor.shutdownNow();
			keepperExecutor.interrupt();
			commandExecutors.interrupt();
		}
	}
	
	/**
	 * 	任务入队
	 * @param task 任务信息
	 * @return
	 */
	@Override
	public boolean enqTask(RetryTask task) {
	    task.setServId(conf.getServiceId());
        return retryQueueDao.insert(task)>0;
	}
	
	/**
	 * 	更新任务
	 * @param task
	 */
	@Override
	public void updateTask(RetryTask task) {
		if(task.getId()==null||task.getScanner()==null) {
			throw new IllegalStateException("更新任务时必须指定id和scanner(只有被锁定状态的任务可以更新)");
		}
		retryQueueDao.update(task);
	}

	
	
	@Override
	public boolean hasRetryHandler(String bizCode) {
		return retryHandlers.containsKey(bizCode);
	}
	
	@Override
	public void registerRetryHandler(String bizCode,RetryHandler handler) {
		retryHandlers.put(bizCode, handler);
	}

	public void setConf(Conf conf) {
		this.conf = conf;
	}

	public void setRetryHandlers(Map<String, RetryHandler> retryHandlers) {
		if(retryHandlers!=null) {
			this.retryHandlers.putAll(retryHandlers);
		}
	}

	public void setRetryQueueDao(RetryQueueDao retryQueueDao) {
		this.retryQueueDao = retryQueueDao;
	}

    public void setCommandWorkerCount(int commandWorkerCount) {
        this.commandWorkerCount = commandWorkerCount;
    }

    @Override
    public void triggeScan(int delayMilles) {
        
        if(!scanner.isRunning()) {
            try {
                //触发实时扫描,降低任务延迟
                scannerExecutor.schedule(scanner, delayMilles, TimeUnit.MILLISECONDS);    
            }catch(Exception e){}
        }
    }

	
}
